#include <pthread.h>
#include <unistd.h>
#include <cstdio>
#include <cstring>
#include <string>
#include <vector>
#include <sys/wait.h>
#include "ServerCommon.h"
#include "AgentCommon.h"
#include "TcpService.h"

static TcpService connectToServer;
static pthread_mutex_t send_lock = PTHREAD_MUTEX_INITIALIZER;

void startMaster();
void startSlave();

void* sendHeartbeat(void *args)
{
	char buf[CMM_MESS_LEN];
	char hostname[CMM_MESS_LEN];
	gethostname(hostname, CMM_MESS_LEN);
	sprintf(buf, "ping");
	connectToServer.TcpSend(buf, CMM_MESS_LEN);
	sprintf(buf, "2 2.4 1 448.0 %s", hostname);
	connectToServer.TcpSend(buf, CMM_MESS_LEN);
	sleep(5);
	while (1) {
		// 发送心跳信息
		pthread_mutex_lock(&send_lock);
		sprintf(buf, "ping");
		connectToServer.TcpSend(buf, CMM_MESS_LEN);
		pthread_mutex_unlock(&send_lock);
		// 睡眠5秒
		sleep(5);
	}
	return NULL;
}

void* recvMessage(void *args)
{
	char buf[CMM_MESS_LEN];
	while (1) {
		// 等待接收信息
		connectToServer.TcpRecv(buf, CMM_MESS_LEN);
		// 处理信息
		if (strncmp(buf, "startMaster", strlen("startMaster")) == 0) {
			startMaster();
		} else if (strncmp(buf, "startSlave", strlen("startSlave")) == 0) {
			startSlave();
		}
	}
	return NULL;
}

void startMaster() {
	std::string mpiFilePath;
	std::string appPath;
	std::string inputPath;
	std::vector<struct keyValue> vec;
	double CGRatio;
	int total, i;
	struct keyValue tmp;
	char mess[CMM_MESS_LEN];
	int pid, status;
	int data_len;
	int agentSocket;
	TcpService connectToAgent;
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	mpiFilePath = mess;
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	appPath = mess;
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	inputPath = mess;
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	sscanf(mess, "%lf", &CGRatio);
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	sscanf(mess, "total %d", &total);
	for (i=0;i<total;i++) {
		connectToServer.TcpRecv(mess, CMM_MESS_LEN);
		sscanf(mess, "%d %d", &tmp.key, &tmp.value);
		vec.push_back(tmp);
	}
	connectToAgent.createService(AGENT_IP, AGENT_PORT, 1);
	if ((pid = fork()) == 0) {
		execlp("mpirun","mpirun",
				"--hostfile", mpiFilePath.c_str(),
				appPath.c_str(), NULL);
	}
	agentSocket = connectToAgent.acceptConnection();
	sprintf(mess, "%s", inputPath.c_str());
	send(agentSocket, mess, CMM_MESS_LEN, 0);
	sprintf(mess, "%lf", CGRatio);
	send(agentSocket, mess, CMM_MESS_LEN, 0);
	sprintf(mess, "total %d", total);
	send(agentSocket, mess, CMM_MESS_LEN, 0);
	for (i=0;i<total;i++) {
		sprintf(mess, "%d %d", vec[i].key, vec[i].value);
		send(agentSocket, mess, CMM_MESS_LEN, 0);
	}
	waitpid(pid, &status, 0);
	recv(agentSocket, mess, CMM_MESS_LEN, 0);
	sscanf(mess, "data %d", &data_len);
	pthread_mutex_lock(&send_lock);
	connectToServer.TcpSend(mess, CMM_MESS_LEN);
	for (i=0;i<data_len;i+=CMM_MESS_LEN) {
		recv(agentSocket, mess, CMM_MESS_LEN, 0);
		connectToServer.TcpSend(mess, CMM_MESS_LEN);
	}
	pthread_mutex_unlock(&send_lock);
	close(agentSocket);
	connectToAgent.closeService();
	return;
}

void startSlave() {
	std::string inputPath;
	std::vector<struct keyValue> vec;
	double CGRatio;
	int total, i;
	struct keyValue tmp;
	char mess[CMM_MESS_LEN];
	int data_len;
	int agentSocket;
	TcpService connectToAgent;
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	inputPath = mess;
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	sscanf(mess, "%lf", &CGRatio);
	connectToServer.TcpRecv(mess, CMM_MESS_LEN);
	sscanf(mess, "total %d", &total);
	for (i=0;i<total;i++) {
		connectToServer.TcpRecv(mess, CMM_MESS_LEN);
		sscanf(mess, "%d %d", &tmp.key, &tmp.value);
		vec.push_back(tmp);
	}
	connectToAgent.createService(AGENT_IP, AGENT_PORT, 1);
	agentSocket = connectToAgent.acceptConnection();
	sprintf(mess, "%s", inputPath.c_str());
	send(agentSocket, mess, CMM_MESS_LEN, 0);
	sprintf(mess, "%lf", CGRatio);
	send(agentSocket, mess, CMM_MESS_LEN, 0);
	sprintf(mess, "total %d", total);
	send(agentSocket, mess, CMM_MESS_LEN, 0);
	for (i=0;i<total;i++) {
		sprintf(mess, "%d %d", vec[i].key, vec[i].value);
		send(agentSocket, mess, CMM_MESS_LEN, 0);
	}
	recv(agentSocket, mess, CMM_MESS_LEN, 0);
	sscanf(mess, "data %d", &data_len);
	pthread_mutex_lock(&send_lock);
	connectToServer.TcpSend(mess, CMM_MESS_LEN);
	for (i=0;i<data_len;i+=CMM_MESS_LEN) {
		recv(agentSocket, mess, CMM_MESS_LEN, 0);
		connectToServer.TcpSend(mess, CMM_MESS_LEN);
	}
	pthread_mutex_unlock(&send_lock);
	close(agentSocket);
	connectToAgent.closeService();
	return;
}

int main()
{
	pthread_t tidpRecv, tidpSend;
	// 建立连接
	connectToServer.connectToService(CMM_IP, CMM_PORT);
	// 开一个线程发送心跳包
	pthread_create(&tidpSend, NULL, sendHeartbeat, NULL);
	// 开一个线程接收启动信息
	pthread_create(&tidpRecv, NULL, recvMessage, NULL);
	// 等待线程结束
	pthread_join(tidpSend, NULL);
	return 0;
}
